package com.atguigu.bigdata.gmall.realtime.app.dws;/**
 * @Classname Dws_03_VideoPlayWindow_Cache_Async
 * @Description TODO
 * @Date 2022/10/26 14:21
 * @Created by 86152
 */

import com.alibaba.fastjson.JSONObject;
import com.atguigu.bigdata.gmall.realtime.app.BaseSQLApp;
import com.atguigu.bigdata.gmall.realtime.bean.EvalBean;
import com.atguigu.bigdata.gmall.realtime.bean.VideoP;
import com.atguigu.bigdata.gmall.realtime.common.Constant;
import com.atguigu.bigdata.gmall.realtime.util.AtguiguUtil;
import com.atguigu.bigdata.gmall.realtime.util.FlinkSinkUtil;
import com.atguigu.bigdata.gmall.realtime.util.SQLUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichReduceFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collection;

/**
 * @author: Lime
 * @description 视频播放
 * @date: 2022/10/26 14:21
 */
public class Dws_03_VideoPlayWindow_Cache_Async extends BaseSQLApp {

    public static void main(String[] args) {
        new Dws_03_VideoPlayWindow_Cache_Async().init(
                3011,
                2,
                "Dws_03_VideoPlayWindow_Cache_Async"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));
        //dwd_video_play

        //3.搞入维度表视频表
        readBaseDicVideo(tEnv);

//        tEnv.sqlQuery("select * from video_info").execute().print();
        //1.创建dwd映射topic表dwd_video_play
        tEnv.executeSql("create table dwd_video_play(\n" +
                " id string,\n" +
                " course_id bigint,\n" +
                " chapter_id bigint,\n" +
                " user_id string,\n" +
                " ts bigint,\n" +
                " `pt` as proctime()\n" +
                ")" + SQLUtil.getKafkaSource(Constant.TOPIC_DWD_VIDEO_PLAY,"Dws_03_VideoPlayWindow_Cache_Async"));

        //测试
//        tEnv.executeSql("select * from dwd_video_play").print();

        //2.创建dwd映射topic表dwd_traffic_appVideo
        tEnv.executeSql("create table dwd_traffic_appVideo(\n" +
                " appVideo map<String, String>,\n" +
                " common map<String, String>,\n" +
                " ts bigint,\n" +
                " `pt` as proctime()\n" +
                ")" + SQLUtil.getKafkaSource(Constant.TOPIC_DWD_TRAFFIC_APPVIDEO,"Dws_03_VideoPlayWindow_Cache_Async"));
//        tEnv.executeSql("select * from dwd_traffic_appVideo").print();

        //3.过滤视频观看数据
        Table VideoSee = tEnv.sqlQuery("select\n" +
                " cast(appVideo['video_id'] as bigint) video_id,\n" +
                " appVideo['play_sec'] play_sec,\n" +
                " common['uid'] uid,\n" +
                " ts,\n" +
                " pt\n" +
                "from dwd_traffic_appVideo");
        tEnv.createTemporaryView("VideoSee",VideoSee);

//        VideoSee.execute().print();

        Table t1 = tEnv.sqlQuery("select\n" +
                " t1.course_id,\n" +
                " t1.chapter_id,\n" +
                " t1.user_id,\n" +
                " t1.ts,\n" +
                " t1.pt,\n" +
                " t2.id video_id,\n" +
                " t2.video_name\n" +
                "from dwd_video_play t1 join video_info for system_time as of t1.pt as t2 on t1.chapter_id = t2.chapter_id");
//        t1.execute().print();
        tEnv.createTemporaryView("t1",t1);

        Table tt = tEnv.sqlQuery("select\n" +
                " '0' stt,\n" +
                " '0' edt,\n" +
                " t1.course_id,\n" +
                " t1.chapter_id,\n" +
                " t1.user_id,\n" +
                " t1.ts,\n" +
                " t1.video_id,\n" +
                " t1.video_name,\n" +
                " t2.play_sec,\n" +
                " 0 video_count,\n" +
                " t1.ts p_uv,\n" +
                " t1.ts all_time\n" +
                "from t1 join VideoSee t2 on t1.video_id = t2.video_id");


        SingleOutputStreamOperator<VideoP> resultStream = tEnv
                .toRetractStream(tt, VideoP.class)
                .filter(t -> t.f0)
                .map(t -> {
                    return t.f1;
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<VideoP>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((bean, ts) -> bean.getTs() * 1000)
                )
                .keyBy(VideoP::getChapter_id)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(
                        new ReduceFunction<VideoP>() {
//                            private MapState<String, Integer> mapState;
//                            @Override
//                            public void open(Configuration parameters) throws Exception {
//                                //写一个去重uid逻辑
//                                mapState = getRuntimeContext().getMapState(new MapStateDescriptor<String, Integer>("map-State", Types.STRING, Types.INT));
//                            }

                            @Override
                            public VideoP reduce(VideoP b1, VideoP b2) throws Exception {
                                b1.setVideo_count(b1.getVideo_count() + b2.getVideo_count() + 1);
                                b1.setAll_time(Long.parseLong(b1.getPlay_sec()) + Long.parseLong(b1.getPlay_sec()));

//                                b1.setP_uv();
//                                if (!mapState.contains(b2.getUser_id())) {
//                                    mapState.put(b2.getUser_id(), 1);
//                                }
//
//
//                                if (!mapState.contains(b2.getUser_id())) {
//                                    b1.setP_uv(b1.getP_uv() + 1);
//                                    mapState.put(b2.getUser_id(), 1);
//                                }
                                return b1;
                            }
                        }
                        , new ProcessWindowFunction<VideoP, VideoP, Long, TimeWindow>() {
                            @Override
                            public void process(Long aLong,
                                                Context ctx,
                                                Iterable<VideoP> elements,
                                                Collector<VideoP> out) throws Exception {

                                VideoP bean = elements.iterator().next();
                                // 补充窗口开始和结束
                                bean.setStt(AtguiguUtil.toDatTime(ctx.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDatTime(ctx.window().getEnd()));

                                // 表示这个结果的统计时间
                                bean.setTs(System.currentTimeMillis());
                                out.collect(bean);

                            }
                        });

//        resultStream.print("sss");



        // 3. 写出到 clickhouse 中
        writeToClickhouse(resultStream);


        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }


//        select
//        count(*),   //视频播放次数
//        count(distinct user_id),   //观看人数
//                sum(play_sec)
//        from t1 join t2 on 时间 and user_id
//        group by chapter_id


    }

    private void writeToClickhouse(SingleOutputStreamOperator<VideoP> resultStream) {
        resultStream
                .addSink(FlinkSinkUtil.getClickHouseSink("dws_video_play_window",
                        VideoP.class
                ));
    }


    private void readBaseDicVideo(StreamTableEnvironment tEnv) {
        tEnv.executeSql("create table video_info ( " +
                "  id bigint, " +
                "  video_name string, " +
                "  chapter_id bigint " +
                ") WITH ( " +
                "  'connector' = 'jdbc', " +
                "  'driver' = 'com.mysql.cj.jdbc.Driver', " +
                "  'url' = 'jdbc:mysql://hadoop162:3306/gmall2022?useSSL=false', " +
                "  'table-name' = 'video_info', " +
                "  'username' = 'root', " +
                "  'password' = 'aaaaaa',  " +
                "  'lookup.cache.max-rows' = '10'," +
                "  'lookup.cache.ttl' = '1 hour' " +
                ")");

        //
    }
}
